package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import com.atguigu.gmall.realtime.app.func.KeywordUDTF;

/**
 * Created by 黄凯 on 2023/7/13 0013 11:00
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 流量域：搜索关键词聚合统计
 * * 需要启动的进程
 * *      zk、kafka、flume、DwdTrafficBaseLogSplit、DwsTrafficSourceKeywordPageViewWindow
 */
public class DwsTrafficSourceKeywordPageViewWindow {

    public static void main(String[] args) {

        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //1.4 注册自定义函数到表执行环境中
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //TODO 2.检查点相关的设置(略)
        env.enableCheckpointing(5000L);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从页面日志中读取数据  创建动态表 指定Watermark生成策略以及提取事件时间字段
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_keyword_group";

        /**
         * >>>>:4> {"common":{"ar":"28","uid":"1728","os":"Android 13.0","ch":"xiaomi","is_new":"0","md":"realme Neo2","mid":"mid_162",
         * "vc":"v2.1.134","ba":"realme","sid":"ce0398d2-26cd-47aa-a761-92710d959deb"},
         * "page":{"page_id":"payment","item":"476920","during_time":10429,"item_type":"order_id","last_page_id":"trade"},"ts":1654647013000}
         *
         * ===============================
         *
         *
         */
        tableEnv.executeSql("create table page_log(\n" +
                "\n" +
                "    common map<string,string>,\n" +
                "    page map<string,string>,\n" +
                "    ts bigint,\n" +
                "    row_time as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
                "    WATERMARK FOR row_time AS row_time\n" +
                "\n" +
                ") " + KafkaUtil.getKafkaDDL(topic, groupId));

//        tableEnv.executeSql("select * from page_log").print();

        //TODO 4.过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
                "    page['item'] fullword,\n" +
                "    row_time\n" +
                "    from page_log\n" +
                "where page['last_page_id'] = 'search'\n" +
                "and page['item_type'] = 'keyword'\n" +
                "and page['item'] is not null");

        tableEnv.createTemporaryView("search_table",searchTable);

//        tableEnv.executeSql("select * from search_table").print();

        //TODO 5.使用自定义函数对搜索内容分词 并和原表字段进行关联
        Table splitTable = tableEnv.sqlQuery("select keyword,\n" +
                "       row_time\n" +
                "from search_table,\n" +
                "     LATERAL TABLE(ik_analyze(fullword)) t(keyword)");

        tableEnv.createTemporaryView("split_table", splitTable);

//        tableEnv.executeSql("select * from split_table").print();

        //TODO 6.分组、开窗、聚合统计
        /**
         * +----+--------------------------------+--------------------------------+--------------------------------+--------------------------------+----------------------+
         * | op |                            stt |                            edt |                        keyword |                       cur_date |        keyword_count |
         * +----+--------------------------------+--------------------------------+--------------------------------+--------------------------------+----------------------+
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                           前端 |                       20220608 |                    2 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                         python |                       20220608 |                    1 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                             大 |                       20220608 |                    6 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                           java |                       20220608 |                    1 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                           数据 |                       20220608 |                    6 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                         hadoop |                       20220608 |                    4 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                         多线程 |                       20220608 |                    2 |
         * | +I |            2022-06-08 11:51:40 |            2022-06-08 11:51:50 |                         数据库 |                       20220608 |                    1 |
         */
        Table result = tableEnv.sqlQuery("select\n" +
                "    DATE_FORMAT(TUMBLE_START(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "    DATE_FORMAT(TUMBLE_END(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt,\n" +
                "    keyword,\n" +
                "    date_format(TUMBLE_START(row_time, INTERVAL '10' SECOND), 'yyyyMMdd') cur_date,\n" +
                "    count(*) keyword_count\n" +
                "    from split_table\n" +
                "group by TUMBLE(row_time,INTERVAL '10' SECOND ),keyword");

        tableEnv.createTemporaryView("res_table",result);

//        tableEnv.executeSql("select * from res_table").print();

        //TODO 7.将聚合的结果写到Doris
        tableEnv.executeSql("CREATE table doris_t(  " +
                " stt string, " +
                " edt string, " +
                " keyword string, " +
                " cur_date string, " +
                " keyword_count bigint " +
                ")WITH (" +
                "  'connector' = 'doris', " +
                "  'fenodes' = 'hadoop102:7030', " +
                "  'table.identifier' = 'gmall.dws_traffic_source_keyword_page_view_window', " +
                "  'username' = 'root', " +
                "  'password' = 'aaaaaa', " +
                "  'sink.properties.format' = 'json', " +
                "  'sink.properties.read_json_by_line' = 'true', " +
                "  'sink.buffer-count' = '4', " +
                "  'sink.buffer-size' = '4086'," +
                "  'sink.enable-2pc' = 'false' " + // 测试阶段可以关闭两阶段提交,方便测试
                ")  ");
        tableEnv.executeSql("insert into doris_t select * from res_table");


    }

}
